package com.ruoyi.logCen.dKafka;


import com.ruoyi.logCen.util.SpringContextUtils;
import com.ruoyi.logCen.util.oConvertUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.config.MethodKafkaListenerEndpoint;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.MessageListener;
import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
import org.springframework.kafka.listener.MessageListenerContainer;
import org.springframework.messaging.handler.annotation.support.DefaultMessageHandlerMethodFactory;
import org.springframework.stereotype.Service;
import java.lang.reflect.Method;
import java.util.List;

/**
 * KafkaListenerService 新
 */
@Service
public class KafkaListenerService {

    @Autowired
    private KafkaListenerEndpointRegistry registry;

    @Autowired
    private KafkaMessageHandlerService kafkaMessageHandlerService;

    public void addListener(String topic, String groupId) throws NoSuchMethodException {
        String id = topic + "-" + groupId;
        if (registry.getListenerContainer(id) != null) {
            System.out.println("Listener already exists for topic: " + topic + " and group: " + groupId);
            return;
        }
        MethodKafkaListenerEndpoint<String, String> endpoint = new MethodKafkaListenerEndpoint<>();
        //设置监听器端点相关信息
        //设置Id
        endpoint.setId(id);
        //设置消费者组
        endpoint.setGroupId(groupId);
        //设置要监听的topic数组，可以是多个
        endpoint.setTopics(topic);
        //设置每个监听器线程数
        endpoint.setConcurrency(1);
        //设置批量监听
        endpoint.setBatchListener(true);
        //设置消息处理工厂类，这里用的是默认工厂
        endpoint.setMessageHandlerMethodFactory(new DefaultMessageHandlerMethodFactory());
        //设置实际处理的Bean对象，即实际的对象，比如new Class();
        endpoint.setBean(kafkaMessageHandlerService);
        //设置实际处理的方法(包含方法名和参数)
        Method method = KafkaMessageHandlerService.class.getDeclaredMethod("handleMessage", List.class);
        endpoint.setMethod(method);
        //注册Container并启动，startImmediately表示立马启动
        registry.registerListenerContainer(endpoint, SpringContextUtils.getBean(KafkaListenerContainerFactory.class), true);
        System.out.println("Added listener for topic: " + topic + " and group: " + groupId);
    }


    public boolean startListener(String listenerId) {
        MessageListenerContainer listenerContainer =
                registry.getListenerContainer(listenerId);
        assert listenerContainer != null : false;

        if(oConvertUtils.isEmpty(listenerContainer)){
            return false;
        }
        if(!listenerContainer.isRunning()){
            listenerContainer.start();
        }

        return true;
    }

    public boolean stopListener(String listenerId) {
        MessageListenerContainer listenerContainer =
                registry.getListenerContainer(listenerId);
        assert listenerContainer != null : false;
        listenerContainer.stop();
        return true;
    }


}
